package com.vodafone.mapreduce.processor;

import java.util.List;
import java.util.Map;

import org.corf.endpoint.collector.Collector;
import org.corf.endpoint.emitter.Emitter;
import org.corf.payload.Payload;
import org.corf.processor.DefaultProcessor;

import com.vodafone.mapreduce.dto.function.MapInput;
import com.vodafone.mapreduce.map.CombineFunction;
import com.vodafone.mapreduce.map.MapFunction;

public class MapInputDtoProcessor<K, V, L, W> extends DefaultProcessor {

    private MapFunction<K, V, L, W> mapFunction;
    private CombineFunction<K, V, L, W> combineFunction;

    public MapInputDtoProcessor() {
        super(null, null, null);
    }

    protected MapInputDtoProcessor(String name, Collector collector, Emitter emitter) {
        super(name, collector, emitter);
    }

    public void setMapFunction(MapFunction<K, V, L, W> mapFunction) {
        this.mapFunction = mapFunction;
    }

    public void setCombineFunction(CombineFunction<K, V, L, W> combineFunction) {
        this.combineFunction = combineFunction;
    }

    public Map<L, List<W>> processMapMessage(K k, V v) {
        mapFunction.initialize(k, v);
        mapFunction.map();
        combineFunction.initialize(mapFunction.getMapResult());
        combineFunction.combine();
        return combineFunction.getCombineResult();
    }

    @Override
    public void process(Payload p) {
        MapInput<K, V> input = (MapInput<K, V>) p;
        mapFunction.initialize(input.getKey(), input.getValue());
        mapFunction.map();
        combineFunction.initialize(mapFunction.getMapResult());
        combineFunction.combine();
        combineFunction.getCombineResult();
    }

    protected void sendReduceInput(Map<L, List<W>> intermediate) {
        /*
         * if (messageProducers != null) { // Send out the crap to JMS. try { ReduceInput<L, W> c = new ReduceInput<L, W>(intermediate);
         * Message m = jmsProducerSessions.get(0).createObjectMessage((Serializable) c); m.setStringProperty("created-by", "MapFunction");
         * messageProducers.get(0).send(m); } catch (JMSException ex) { throw new RuntimeException("Failed to send out data to JMS!", ex); }
         * }
         */
    }

}
